package com.cl.ks.flow.handler;

import com.alibaba.fastjson.JSONObject;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.cl.ks.entity.SysSourceMappingTableField;
import com.cl.ks.flow.base.BaseFlowNodeHandler;
import com.cl.ks.flow.enums.NodeHandlerEnum;
import com.cl.ks.utils.DbUtil;
import com.jandar.ds.pojo.DbInfo;
import com.querydsl.jpa.impl.JPAQueryFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

import static com.cl.ks.constant.QTableConstant.*;

@Component
public class DataSourceHandler extends BaseFlowNodeHandler {
    private final DbUtil dbUtil;
    private final JPAQueryFactory queryFactory;


    public DataSourceHandler(DbUtil dbUtil, JPAQueryFactory queryFactory) {
        this.dbUtil = dbUtil;
        this.queryFactory = queryFactory;
    }

    @Override
    public SparkResult process(ProcessParam processParam) {
        JSONObject expression = processParam.getSparkNode().getNodeExpression();
        String tableName = String.valueOf(expression.get("targetTable"));
        String dbCode = String.valueOf(expression.get("dbCode"));
        Integer sourceConfigId = queryFactory.select(qSysSourceConfig.id).from(qSysSourceConfig).where(qSysSourceConfig.targetTable.eq(tableName).and(qSysSourceConfig.dbCode.eq(dbCode))).fetchFirst();
        if (sourceConfigId == null) {
            throw new RuntimeException(dbCode + ":" + tableName + " 数据源不存在");
        }
        processParam.getKsFlowGraph().getSourceConfigIdList().add(String.valueOf(sourceConfigId));
        List<SysSourceMappingTableField> sysSourceMappingTableFields = queryFactory.selectFrom(qSysSourceMappingTableField).where(qSysSourceMappingTableField.tableName.eq(tableName).and(qSysSourceMappingTableField.dbCode.eq(dbCode))).fetch();
        List<String> existFields = dbUtil.listFieldName(dbCode, tableName);

        List<String> fieldsAs = new ArrayList<>();

        List<String> fields = new ArrayList<>();
        List<String> fieldsAlias = new ArrayList<>();

        sysSourceMappingTableFields.stream().filter(field -> !existFields.contains(field.getTableField())).forEach(sysSourceMappingTableField -> {
            String tableField = sysSourceMappingTableField.getTableField();
            String asName = sysSourceMappingTableField.getTableField();
            if (StringUtils.hasLength(sysSourceMappingTableField.getKsDataFieldCode())) {
                asName = sysSourceMappingTableField.getKsDataFieldCode();
            }
            String finalAsName = asName;
            Optional<String> exist = fieldsAlias.stream().filter(finalAsName::equalsIgnoreCase).findFirst();
            if (exist.isPresent()) {
                asName = asName + "_repeat";
            }
            fields.add(tableField);
            fieldsAlias.add(asName.trim());

        });
        for (int i = 0; i < fields.size(); i++) {
            fieldsAs.add(fields.get(i) + " as " + fieldsAlias.get(i));
        }

        // 输出数据
        Integer lastPosition = 0;
        if (processParam.getKsFlowGraph().getIncrementFlag() != null && processParam.getKsFlowGraph().getIncrementFlag()) {
            lastPosition = queryFactory.select(qKsFlowGraphLog.lastPosition).from(qKsFlowGraphLog).where(qKsFlowGraphLog.ksFlowGraphId.eq(processParam.getKsFlowGraph().getId()).and(qKsFlowGraphLog.sparkNodeCode.eq(processParam.getSparkNode().getCode()))).orderBy(qKsFlowGraphLog.ksFlowGraphRecordId.desc()).fetchFirst();
            if (lastPosition == null) {
                lastPosition = 0;
            }
        }
        Long count = Long.parseLong(String.valueOf(dbUtil.fetchSQL(dbCode, "select count(1) as count from " + tableName).get(0).get("count")));

        processParam.getKsFlowGraph().setSourceConfigDataTotal((int) (processParam.getKsFlowGraph().getSourceConfigDataTotal() + (count - lastPosition)));

        processParam.getKsFlowGraphLog().setLastPosition(Math.toIntExact(count));

        long limit = count - lastPosition;
        if (processParam.getTestFlag() == true || StringUtils.hasLength(processParam.getKsFlowGraph().getFailedNodeCode())) {
            limit = 0;
        }

        DbInfo dbInfo = dbUtil.getDbInfo(dbCode);
        SparkParam sparkParam = new SparkParam();
        expression.put("limit", limit);
        expression.put("skip", lastPosition);
        expression.put("url", dbInfo.getUrl());
        expression.put("username", dbInfo.getUsername());
        expression.put("password", dbInfo.getPassword());
        expression.put("tableName", tableName);
        expression.put("fieldsAs", String.join(",", fieldsAs));
        sparkParam.setNodeExpression(expression);
        SparkResult sparkResult = sparkNodeFactory.getSparkNodeByCode("dataSource").handle(sparkParam);
        return sparkResult;
    }
//    @Override
//    public SparkResult process(BaseFlowNodeHandler.ProcessParam processParam) {
////        JSONArray res = new JSONArray();
//        JSONObject expression = processParam.getSparkNode().getNodeExpression();
//        String sparkNodeName = "";
//        String tableName = String.valueOf(expression.get("targetTable"));
//        String dbCode = String.valueOf(expression.get("dbCode"));
//        sparkNodeName = "数据源：" + dbCode + "-" + tableName;
////        setProgress(progressKey, sparkNodeCode, sparkNodeName + " 字段查询", 0, 0);
//        Integer sourceConfigId = queryFactory.select(qSysSourceConfig.id).from(qSysSourceConfig).where(qSysSourceConfig.targetTable.eq(tableName).and(qSysSourceConfig.dbCode.eq(dbCode))).fetchFirst();
//        if (sourceConfigId == null) {
//            throw new RuntimeException(dbCode + ":" + tableName + " 数据源不存在");
//        }
//        processParam.getKsFlowGraph().getSourceConfigIdList().add(String.valueOf(sourceConfigId));
//        List<SysSourceMappingTableField> sysSourceMappingTableFields = queryFactory.selectFrom(qSysSourceMappingTableField).where(qSysSourceMappingTableField.tableName.eq(tableName).and(qSysSourceMappingTableField.dbCode.eq(dbCode))).fetch();
//
//        // 用于插入数据源数据映射
//        Map<String, String> fieldsMap = new HashMap<>();
//        // 用于查询数据源输出映射
//        List<String> fields = new ArrayList<>();
//        sysSourceMappingTableFields.forEach(sysSourceMappingTableField -> {
//            String tableField = sysSourceMappingTableField.getTableField();
//            String asName = sysSourceMappingTableField.getTableField();
//            if (StringUtils.hasLength(sysSourceMappingTableField.getKsDataFieldCode())) {
//                asName = sysSourceMappingTableField.getKsDataFieldCode();
//            }
//            fieldsMap.put(tableField, asName);
//            fields.add(tableField + " as " + asName);
//        });
//
//        // 数据输入到数据源
//        Map<String, JSONArray> inputDataList = processParam.getSparkNode().getInputDataList();
//        int total = 0;
//        for (JSONArray jsonArray : inputDataList.values()) {
//            total += jsonArray.size();
//        }
//        inputDataList.forEach((nodeName, data) -> {
//            for (int i = 0; i < data.size(); i++) {
//                Map<String, String> insertItem = new TreeMap<>();
//                JSONObject item = data.getJSONObject(i);
//                item.entrySet().stream().filter(entry -> fieldsMap.containsKey(entry.getKey())).forEach(entry -> {
//                    insertItem.put(fieldsMap.get(entry.getKey()), item.getString(entry.getKey()));
//                });
//                dbUtil.insert(dbCode, tableName, insertItem);
//            }
//        });
////        Map<String, String> dataHdfsPathList = sparkNode.getDataHdfsPathList();
////        AtomicInteger total = new AtomicInteger(0);
////
////        dataHdfsPathList.forEach((nodeName, dataHdfsPath) -> {
////            JSONArray data = sparkUtil.readJSONArray(dataHdfsPath);
////            total.addAndGet(data.size());
////            for (int i = 0; i < data.size(); i++) {
////                Map<String, String> insertItem = new TreeMap<>();
////                JSONObject item = data.getJSONObject(i);
////                item.entrySet().stream().filter(entry -> fieldsMap.containsKey(entry.getKey())).forEach(entry -> {
////                    insertItem.put(fieldsMap.get(entry.getKey()), item.getString(entry.getKey()));
////                });
////                dbUtil.insert(dbCode, tableName, insertItem);
////            }
////        });
//
//
//        // 输出数据
//        Integer lastPosition = 0;
//        if (processParam.getKsFlowGraph().getIncrementFlag() != null && processParam.getKsFlowGraph().getIncrementFlag()) {
//            lastPosition = queryFactory.select(qKsFlowGraphLog.lastPosition).from(qKsFlowGraphLog).where(qKsFlowGraphLog.ksFlowGraphId.eq(processParam.getKsFlowGraph().getId()).and(qKsFlowGraphLog.sparkNodeCode.eq(processParam.getSparkNode().getCode()))).orderBy(qKsFlowGraphLog.ksFlowGraphRecordId.desc()).fetchFirst();
//            if (lastPosition == null) {
//                lastPosition = 0;
//            }
//        }
//        Long count = Long.parseLong(String.valueOf(dbUtil.fetchSQL(dbCode, "select count(1) as count from " + tableName).get(0).get("count")));
//        List<Map<String, Object>> dataList = new ArrayList<>();
//        if (count - lastPosition > 0) {
//            String limit = " limit " + lastPosition + " , " + (count - lastPosition);
//            dataList = dbUtil.fetchSQL(dbCode, "select " + String.join(",", fields) + " from " + tableName + limit);
//        }
////        setProgress(progressKey, sparkNodeCode, sparkNodeName + " 数据输出", dataList.size(), dataList.size());
//
//
//        dataList.forEach(item -> {
//
//            item.put("DB_CODE", dbCode);
//            item.put("TABLE_NAME", tableName);
////            item.put("DB_CODE", dbCode);
////            item.put("TABLE_NAME", tableName);
//        });
//
//        processParam.getKsFlowGraph().setSourceConfigDataTotal(processParam.getKsFlowGraph().getSourceConfigDataTotal() + dataList.size());
//
////        String hdfsPath = "/temp/"+ ksFlowGraphRecord.getId() + sparkNodeName + sparkNodeCode +".json";
////        sparkUtil.saveJSONArray(hdfsPath, JSONArray.parseArray(JSON.toJSONString(dataList)));
////        sparkResult.setHdfsPath(hdfsPath);
////        res.addAll(dataList);
////        sparkResult.setTotal(total.get());
//
//        processParam.getKsFlowGraphLog().setLastPosition(Math.toIntExact(count));
//
//        JSONArray fieldList = JSONArray.parseArray(JSONArray.toJSONString(sysSourceMappingTableFields));
//        return SparkResult.success(JSONArray.parseArray(JSON.toJSONString(dataList)), fieldList);
//    }

    @Override
    public NodeHandlerEnum getType() {
        return NodeHandlerEnum.DATA_SOURCE;
    }
}
